package drds.data_migrate.job.read_and_write;

import drds.common.$;
import drds.data_migrate.util.Jdbc;
import drds.data_migrate.util.Updates;
import drds.data_migrate.util.Writes;
import drds.data_migrate.vo.ConnectionInfo;
import drds.data_migrate.vo.JobConfig;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;

/**
 * Created by ibm on 2018/11/3.
 */
@Slf4j
public class WriteJob
{

    @Setter
    @Getter
    private JobConfig jobConfig;

    public WriteJob()
    {


    }


    public WriteJob(JobConfig jobConfig)
    {

        this.jobConfig = jobConfig;
    }

    public void init() throws SQLException
    {
        Writes.doPretreatment(null);


    }

    /*目前只支持MySQL Writer跟Oracle AbstractWriter;检查PreSQL跟PostSQL语法以及insert，delete权限*/
    public void preCheck()
    {
        /*检查PreSql跟PostSql语句*/
        prePostSqlValid();

    }

    public void prePostSqlValid()
    {
        /*检查PreSql跟PostSql语句*/
        Updates.preCheckPrePareSQL();
        Updates.preCheckPostSQL();
    }


    // 一般来说，是需要推迟到 wrapper 中进行pre 的执行（单表情况例外）
    public void prepare() throws SQLException
    {
        int tableNumber = jobConfig.getTableNumberMark();
        if (tableNumber == 1)
        {
            //String username = null;
            //String password = null;

            List<ConnectionInfo> connectionInfoList = this.jobConfig.getConnectionInfoList();
            ConnectionInfo connectionInfo = connectionInfoList.get(0);

            // 这里的 jdbcUrl 已经 append 了合适后缀参数
            //String jdbcUrl = connectionInfo.getJdbcUrl();
            //taskConfig.set(Key.jdbcUrl, jdbcUrl);

            String tableName = connectionInfo.getTableNameList().get(0);
            //taskConfig.set(Key.tableNameList, tableName);

            List<String> preSqlList = connectionInfo.getPreSqlList();//taskConfig.getList(Key.preSqlList, String.class);
            List<String> sqlList = Updates.replaceTableName(preSqlList, tableName);

            //taskConfig.remove(Constant.connection);
            if ($.isHasData(sqlList))
            {
                // 说明有 preSql 配置，则此处删除掉
                //taskConfig.remove(Key.preSqlList);

                Connection connection = Jdbc.getConnection(
                        connectionInfo.getJdbcUrl(), connectionInfo.getUsername(), connectionInfo.getPassword());
                log.info("Begin to execute preSqlList:[{}]. context info:{}.",
                        StringUtils.join(sqlList, ";"), connectionInfo.getJdbcUrl());

                Updates.executeSqlList(connection, sqlList);
                Jdbc.close(null, null, connection);
            }
        }


    }

    /**
     * 切分任务。<br>
     *
     * @param mandatoryNumber 为了做到Reader、Writer任务数对等，这里要求Writer插件必须按照源端的切分数进行切分。否则框架报错！
     */

    public List<TaskConfig> split(
            int mandatoryNumber)
    {
        return Updates.doSplit(mandatoryNumber);
    }

    // 一般来说，是需要推迟到 wrapper 中进行post 的执行（单表情况例外）
    public void post() throws SQLException
    {
        int tableNumber = 1;
        if (tableNumber == 1)
        {


            List<String> postSqlList = this.jobConfig.getPostSqlList();
            List<String> sqlList = Updates.replaceTableName(
                    postSqlList, this.jobConfig.getTableName());

            if ($.isHasData(sqlList))
            {
                // 说明有 postSqlList 配置，则此处删除掉
                //taskConfig.remove(Key.postSqlList);

                Connection connection = Jdbc.getConnection(
                        jobConfig.getJdbcUrl(), jobConfig.getUsername(), jobConfig.getPassword());

                log.info(
                        "Begin to execute postSqlList:[{}]. context info:{}.",
                        StringUtils.join(sqlList, ";"), jobConfig.getJdbcUrl());
                Updates.executeSqlList(connection, sqlList);
                Jdbc.close(null, null, connection);
            }
        }
    }

    public void destroy()
    {
    }

}
